package mq

import (
	"fmt"
	"gin_easy_starter/configs"
	"github.com/IBM/sarama"
	"log"
	"strings"
)

func CreateKafkaTopic(topic string, partitions int32, replicationFactor int16) error {
	config := sarama.NewConfig()
	client, err := sarama.NewClient(configs.Get().Kafka.BrokerList, config)
	if err != nil {
		return fmt.Errorf("Failed to connect to Kafka brokers: %v", err)
	}
	defer client.Close()

	adminClient, err := sarama.NewClusterAdminFromClient(client)
	if err != nil {
		return fmt.Errorf("Failed to get admin client: %v", err)
	}
	defer adminClient.Close()

	topicDetail := &sarama.TopicDetail{
		NumPartitions:     partitions,
		ReplicationFactor: replicationFactor,
	}
	topics, err := adminClient.ListTopics()
	if err != nil {
		return err
	}
	for s := range topics {
		if strings.EqualFold(s, topic) {
			return nil
		}

	}
	err = adminClient.CreateTopic(topic, topicDetail, false) // 第三个参数为 true 则会立即返回结果，false 则会等待完成
	if err != nil {
		return fmt.Errorf("Failed to create topic '%s': %v", topic, err)
	}

	log.Println("Successfully created topic: ", topic)
	return nil
}
